package com.example.flink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.ArrayList;
import java.util.Properties;

/**
 * Created with IntelliJ IDEA.
 * ClassName: FlinkKafkaProducer
 * Package: com.example.flink
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-08-02
 * Time: 21:35
 */

//kafka集成 flink 系统有重名的FlinkKafkaProduce
public class FlinkKafkaProducerDemo {

    public static void main(String[] args) throws Exception {

        //1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置分区数 对应的主题有几个分区 就设置几个分区数据
        env.setParallelism(3);

        //2.准备数据源
        ArrayList<String> wordList = new ArrayList<>();
        wordList.add("hello");
        wordList.add("java");
        DataStreamSource<String> stringDataStreamSource = env.fromCollection(wordList);

        //需要一个kafka生成者 准备一个生产者
        Properties properties = new Properties();
        //必须进行配置 不然还不知道连接哪个集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
        //在kafka创建的生产者  三个参数 主题 序列化器 直接new封装好的 配置参数
        FlinkKafkaProducer<String> first = new FlinkKafkaProducer<>("first", new SimpleStringSchema(), properties);

        //3.添加数据源 输出到kafka生产者
        stringDataStreamSource.addSink(first);

        //4.执行
        env.execute();

    }

}
